package org.lemon.transport.context.service;

import com.google.protobuf.ByteString;
import lombok.extern.slf4j.Slf4j;
import org.lemon.transport.context.auth.GetOrCreateDeviceFromGatewayResponse;
import org.lemon.transport.context.auth.TransportDeviceInfo;
import org.lemon.transport.context.auth.ValidateDeviceCredentialsResponse;
import org.lemon.transport.context.data.device.*;
import org.lemon.transport.context.data.tenant.TenantId;
import org.lemon.transport.context.service.request.DeviceAuthRequestTemplate;
import org.lemon.transport.context.service.discovery.ServiceInfoProvider;
import org.lemon.transport.context.service.queue.*;
import org.lemon.transport.context.data.queue.message.ProtoQueueMessage;
import org.lemon.transport.context.service.queue.provider.NotificationsTopicService;
import org.lemon.transport.context.service.queue.provider.RabbitMqTransportQueueFactory;
import org.lemon.transport.context.service.request.DeviceTransportRequestTemplate;
import org.lemon.transport.context.utils.DefaultThreadFactory;
import org.lemon.transport.context.utils.StringUtils;
import org.server.transport.gen.SessionProto.*;
import org.server.transport.gen.TransportApiProto;
import org.server.transport.gen.TransportProto;
import org.server.transport.gen.TransportProto.*;
import org.lemon.transport.context.model.SessionActivityData;
import org.lemon.transport.context.model.SessionMetaData;
import org.lemon.transport.context.service.event.DeviceDeletedEvent;
import org.lemon.transport.context.service.event.DeviceProfileUpdatedEvent;
import org.lemon.transport.context.service.event.DeviceUpdatedEvent;
import org.lemon.transport.context.session.SessionMsgListener;
import org.lemon.transport.context.utils.TransportForkJoinWorkerThreadFactory;
import org.server.transport.gen.TransportProto;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Service
public class DefaultTransportService implements TransportService, ApplicationEventPublisherAware {

    public static final String SESSION_EXPIRED_MESSAGE = "Session has expired due to last activity time!";
    public static final SessionEventMsg SESSION_EVENT_MSG_OPEN = getSessionEventMsg(SessionEvent.OPEN);
    public static final SessionEventMsg SESSION_EVENT_MSG_CLOSED = getSessionEventMsg(SessionEvent.CLOSED);
    public static final SessionCloseNotificationProto SESSION_CLOSE_NOTIFICATION_PROTO = SessionCloseNotificationProto.newBuilder()
            .setMessage(SESSION_EXPIRED_MESSAGE).build();

    public static final TransportProto.SubscribeToAttributeUpdatesMsg SUBSCRIBE_TO_ATTRIBUTE_UPDATES_ASYNC_MSG = TransportProto.SubscribeToAttributeUpdatesMsg.newBuilder()
            .setSessionType(SessionType.ASYNC).build();
    public static final TransportProto.SubscribeToRPCMsg SUBSCRIBE_TO_RPC_ASYNC_MSG = TransportProto.SubscribeToRPCMsg.newBuilder()
            .setSessionType(SessionType.ASYNC).build();

    private final AtomicInteger atomicTs = new AtomicInteger(0);

    @Value("${transport.sessions.inactivity_timeout}")
    private long sessionInactivityTimeout;
    @Value("${transport.sessions.report_timeout}")
    private long sessionReportTimeout;
    @Value("${queue.transport.poll_interval}")
    private int notificationsPollDuration;

    @Autowired
    private ServiceInfoProvider serviceInfoProvider;
    @Autowired
    private NotificationsTopicService notificationsTopicService;
    @Autowired
    private DeviceAuthRequestTemplate deviceAuthRequestTemplate;
    @Autowired
    private DeviceTransportRequestTemplate deviceTransportRequestTemplate;
    @Autowired
    private RabbitMqTransportQueueFactory rabbitMqTransportQueueFactory;
    @Autowired
    private TransportDeviceProfileCache transportDeviceProfileCache;

    private final SchedulerComponent scheduler;
    public final ConcurrentMap<UUID, SessionMetaData> sessions = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<UUID, SessionActivityData> sessionsActivity = new ConcurrentHashMap<>();


    private QueueConsumer<ProtoQueueMessage<ToTransportMsg>> transportNotificationsConsumer;
    private ExecutorService transportCallbackExecutor;
    private ExecutorService mainConsumerExecutor;
    private volatile boolean stopped = false;

    private ApplicationEventPublisher eventPublisher;

    public DefaultTransportService(SchedulerComponent scheduler) {
        this.scheduler = scheduler;
    }

    @PostConstruct
    public void init() {

        this.transportCallbackExecutor = new ForkJoinPool(20, new TransportForkJoinWorkerThreadFactory(this.getClass().getName()),
                null, true);
        scheduler.scheduleAtFixedRate(this::checkInactivityAndReportActivity, new Random().nextInt((int) sessionReportTimeout), sessionReportTimeout, TimeUnit.MILLISECONDS);
        scheduler.scheduleAtFixedRate(this::invalidateRateLimits, new Random().nextInt((int) sessionReportTimeout), sessionReportTimeout, TimeUnit.MILLISECONDS);
        this.transportNotificationsConsumer = rabbitMqTransportQueueFactory.createTransportNotificationsConsumer();
        TopicPartitionInfo topic = notificationsTopicService.getNotificationsTopic(ServiceType.TRANSPORT, serviceInfoProvider.getServiceId());
        this.transportNotificationsConsumer.subscribe(Collections.singleton(topic));
        this.mainConsumerExecutor = Executors.newSingleThreadExecutor(DefaultThreadFactory.forName("transport-consumer"));

    }

    @AfterStartUp(order = AfterStartUp.TRANSPORT_SERVICE)
    private void start() {
        log.info("main consumer executor run...");
        mainConsumerExecutor.execute(() -> {
            while (!stopped) {
                try {
                    List<ProtoQueueMessage<ToTransportMsg>> records = transportNotificationsConsumer.poll(notificationsPollDuration);
                    if (records.size() == 0) {
                        continue;
                    }
                    records.forEach(record -> {
                        try {
                            if (record != null){
                                processToTransportMsg(record.getValue());
                            }
                        } catch (Throwable e) {
                            log.warn("Failed to process the notification.", e);
                        }
                    });
                    transportNotificationsConsumer.commit();
                } catch (Exception e) {
                    if (!stopped) {
                        log.warn("Failed to obtain message form queue.", e);
                    }
                    try {
                        Thread.sleep(notificationsPollDuration);
                    } catch (InterruptedException e2) {
                        log.trace("Failed to wait until the server has capacity to handle new requests", e2);
                    }
                }
            }
        });
    }

    @Override
    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {

        this.eventPublisher = applicationEventPublisher;
    }

    @Override
    public ExecutorService getCallbackExecutor() {
        return this.transportCallbackExecutor;
    }

    @Override
    public List<GetQueueRoutingInfoResponseMsg> getQueueRoutingInfo(GetAllQueueRoutingInfoRequestMsg msg) {
        return null;
    }

    /**
     * 设备accessToken 验证
     *
     * @param transportType
     * @param token
     * @param responseCallback
     */
    @Override
    public void deviceTokenValidate(DeviceTransportType transportType, String token,
                                    TransportAuthServiceCallback<ValidateDeviceCredentialsResponse> responseCallback) {
        ValidateDeviceTokenRequestMsg.Builder authMsgBuilder = ValidateDeviceTokenRequestMsg.newBuilder();
        authMsgBuilder.setToken(token);
        try {
            ValidateDeviceCredentialsResponseMsg responseMsg = deviceAuthRequestTemplate.deviceTokenValidate(authMsgBuilder.build());
            ValidateDeviceCredentialsResponse result = processValidateResponseMsg(transportType, responseMsg);
            responseCallback.onSuccess(result);
        } catch (Exception e) {
            log.error("deviceTokenValidate", e);
            responseCallback.onError(e);
        }
    }

    /**
     * mqtt 账号密码连接验证
     *
     * @param transportType
     * @param clientId
     * @param username
     * @param password
     * @param responseCallback
     */
    @Override
    public void basicMqttCredValidate(DeviceTransportType transportType, String clientId, String username, String password,
                                      TransportAuthServiceCallback<ValidateDeviceCredentialsResponse> responseCallback) {
        ValidateBasicMqttCredRequestMsg.Builder authMsgBuilder = ValidateBasicMqttCredRequestMsg.newBuilder();
        authMsgBuilder.setClientId(clientId);
        authMsgBuilder.setUserName(username);
        authMsgBuilder.setPassword(password);
        try {
            ValidateDeviceCredentialsResponseMsg responseMsg = deviceAuthRequestTemplate.basicMqttCredTokenValidate(authMsgBuilder.build());
            ValidateDeviceCredentialsResponse result = processValidateResponseMsg(transportType, responseMsg);
            responseCallback.onSuccess(result);
        } catch (Exception e) {
            log.error("basicMqttCredTokenValidate", e);
            responseCallback.onError(e);
        }
    }

    /**
     * 设备x509 证书验证
     *
     * @param transportType
     * @param certHash
     * @param responseCallback
     */
    @Override
    public void deviceX509CertValidate(DeviceTransportType transportType, String certHash,
                                       TransportAuthServiceCallback<ValidateDeviceCredentialsResponse> responseCallback) {
        ValidateDeviceX509CertRequestMsg.Builder authMsgBuilder = ValidateDeviceX509CertRequestMsg.newBuilder();
        authMsgBuilder.setHash(certHash);
        try {
            ValidateDeviceCredentialsResponseMsg responseMsg = deviceAuthRequestTemplate.deviceX509CertValidate(authMsgBuilder.build());
            ValidateDeviceCredentialsResponse result = processValidateResponseMsg(transportType, responseMsg);
            responseCallback.onSuccess(result);
        } catch (Exception e) {
            log.error("deviceX509CertValidate", e);
            responseCallback.onError(e);
        }
    }

    @Override
    public void process(SessionInfoProto sessionInfo, SessionEventMsg msg, TransportServiceCallback<Void> callback) {

        log.info("process device SessionEventMsg... {}", msg.toString());
        sendToRuleEngine(sessionInfo, null, null);
    }

    /**
     * 设备遥测数据处理
     *
     * @param sessionInfo
     * @param msg
     * @param callback
     */
    @Override
    public void process(SessionInfoProto sessionInfo, PostTelemetryMsg msg, TransportServiceCallback<Void> callback) {

        log.info("process device PostTelemetryMsg... {}", msg.toString());
        int dataPoints = 0;
        for (TsKeyValueListProto tsKvProto : msg.getTsKvListList()) {
            dataPoints += tsKvProto.getKvCount();
        }
        reportActivityInternal(sessionInfo);
        for (TsKeyValueListProto tsKv : msg.getTsKvListList()) {
            sendToRuleEngine(sessionInfo, null, null);
        }
        callback.onSuccess(null);
    }

    /**
     * 设备属性数据处理
     *
     * @param sessionInfo
     * @param msg
     * @param callback
     */
    @Override
    public void process(SessionInfoProto sessionInfo, PostAttributeMsg msg, TransportServiceCallback<Void> callback) {

        log.info("process device PostAttributeMsg... {}", msg.toString());
    }

    @Override
    public void process(SessionInfoProto session, GetAttributeRequestMsg requestMsg, TransportServiceCallback<GetAttributeResponseMsg> callback) {


        KeyValueProto key1 = KeyValueProto.newBuilder().setKey("key1").setType(KeyValueType.STRING_V).setStringV("KEY1_VALUE").build();
        KeyValueProto key2 = KeyValueProto.newBuilder().setKey("key2").setType(KeyValueType.DOUBLE_V).setDoubleV(3.1415926).build();
        GetAttributeResponseMsg.Builder builder = GetAttributeResponseMsg.newBuilder();
        builder.addClientAttributeList(0, TsKeyValueProto.newBuilder().setKv(key1));
        builder.addClientAttributeList(1, TsKeyValueProto.newBuilder().setKv(key2));
        builder.setSharedStateMsg(false);
        callback.onSuccess(builder.build());
    }

    @Override
    public void process(ProvisionDeviceRequestMsg requestMsg, TransportServiceCallback<ProvisionDeviceResponseMsg> callback) {

        try {


        } catch (Exception e) {
            log.error("Process ProvisionDeviceRequestMsg", e);
            callback.onError(e);
        }

    }

    /**
     * 处理固件/软件Ota包信息
     *
     * @param sessionInfoProto
     * @param requestMsg
     * @param callback
     */
    @Override
    public void process(SessionInfoProto sessionInfoProto, GetOtaPackageRequestMsg requestMsg, TransportServiceCallback<GetOtaPackageResponseMsg> callback) {
        log.info("process GetOtaPackageRequestMsg...");
        try {

        } catch (Exception e) {
            log.error("Process ProvisionDeviceRequestMsg", e);
            callback.onError(e);
        }
    }

    /**
     * RPC订阅消息
     *
     * @param sessionInfo
     * @param msg
     * @param callback
     */
    @Override
    public void process(SessionInfoProto sessionInfo, SubscribeToRPCMsg msg, TransportServiceCallback<Void> callback) {

    }

    /**
     * 订阅属性更新消息
     *
     * @param sessionInfo
     * @param msg
     * @param callback
     */
    @Override
    public void process(SessionInfoProto sessionInfo, SubscribeToAttributeUpdatesMsg msg, TransportServiceCallback<Void> callback) {

    }

    /**
     * 处理订阅信息
     *
     * @param sessionInfo
     * @param subscriptionInfo
     * @param getAttributeResponseMsgTransportServiceCallback
     */
    @Override
    public void process(SessionInfoProto sessionInfo, SubscriptionInfoProto subscriptionInfo, TransportServiceCallback<GetAttributeResponseMsg> getAttributeResponseMsgTransportServiceCallback) {
    }

    @Override
    public void process(TransportApiProto.GetOrCreateDeviceFromGatewayRequestMsg msg, TransportServiceCallback<GetOrCreateDeviceFromGatewayResponse> callback) {

        GetOrCreateDeviceFromGatewayResponse response = new GetOrCreateDeviceFromGatewayResponse();
        TransportDeviceInfo transportDeviceInfo = new TransportDeviceInfo();
        transportDeviceInfo.setDeviceId(100001L);
        transportDeviceInfo.setDeviceName(msg.getDeviceName());
        transportDeviceInfo.setDeviceType(msg.getDeviceType());
        transportDeviceInfo.setTenantId(TenantId.SYS_TENANT_ID);
        transportDeviceInfo.setProjectId(1L);
        transportDeviceInfo.setDeviceProfileId(1L);
        response.setDeviceInfo(transportDeviceInfo);
        DeviceProfile deviceProfile = new DeviceProfile();
        deviceProfile.setId(1L);
        deviceProfile.setTenantId(TenantId.SYS_TENANT_ID);
        deviceProfile.setTransportType(DeviceTransportType.MQTT);
        callback.onSuccess(response);
    }

    @Override
    public void process(TransportToDeviceActorMsg msg, TransportServiceCallback<Void> callback) {

    }

    @Override
    public void process(SessionInfoProto sessionInfo, ClaimDeviceMsg claimDeviceMsg, TransportServiceCallback<Void> pubAckCallback) {

    }

    @Override
    public void reportActivity(SessionInfoProto sessionInfo) {
        reportActivityInternal(sessionInfo);
    }


    public void log(SessionInfoProto session, String message) {

    }

    @Override
    public SessionMetaData registerAsyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener) {
        return sessions.computeIfAbsent(toSessionId(sessionInfo), (x) -> {
            return new SessionMetaData(sessionInfo, SessionType.ASYNC, listener);
        });
    }

    @Override
    public SessionMetaData registerSyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener, long timeout) {

        SessionMetaData currentSession = new SessionMetaData(sessionInfo, SessionType.SYNC, listener);
        UUID sessionId = toSessionId(sessionInfo);
        sessions.putIfAbsent(sessionId, currentSession);

        SessionCloseNotificationProto notification = SessionCloseNotificationProto.newBuilder().setMessage("session timeout!").build();
        ScheduledFuture<?> executorFuture = scheduler.schedule(() -> {
            listener.onRemoteSessionCloseCommand(sessionId, notification);
        }, timeout, TimeUnit.MILLISECONDS);
        currentSession.setScheduledFuture(executorFuture);
        return currentSession;
    }

    @Override
    public void deregisterSession(SessionInfoProto sessionInfo) {

        UUID sessionId = toSessionId(sessionInfo);
        SessionMetaData currentSession = sessions.get(sessionId);
        if (currentSession != null && currentSession.hasScheduledFuture()) {
            log.debug("Stopping scheduler to avoid resending response if request has been ack.");
            currentSession.getScheduledFuture().cancel(false);
        }
        sessions.remove(sessionId);
    }

    protected void processToTransportMsg(ToTransportMsg toSessionMsg) {
        UUID sessionId = new UUID(toSessionMsg.getSessionIdMSB(), toSessionMsg.getSessionIdLSB());
        SessionMetaData sessionMetaData = sessions.get(sessionId);
        if (sessionMetaData != null) {
            log.trace("[{}] Processing notification: {}", sessionId, toSessionMsg);
            SessionMsgListener listener = sessionMetaData.getListener();
            transportCallbackExecutor.submit(() -> {
                if (toSessionMsg.hasGetAttributesResponse()) {
                    listener.onGetAttributesResponse(toSessionMsg.getGetAttributesResponse());
                }
                if (toSessionMsg.hasAttributeUpdateNotification()) {
                    listener.onAttributeUpdate(sessionId, toSessionMsg.getAttributeUpdateNotification());
                }
                if (toSessionMsg.hasSessionCloseNotification()) {
                    listener.onRemoteSessionCloseCommand(sessionId, toSessionMsg.getSessionCloseNotification());
                }
                if (toSessionMsg.hasToTransportUpdateCredentialsNotification()) {
                    listener.onToTransportUpdateCredentials(toSessionMsg.getToTransportUpdateCredentialsNotification());
                }
                if (toSessionMsg.hasToDeviceRequest()) {
                    listener.onToDeviceRpcRequest(sessionId, toSessionMsg.getToDeviceRequest());
                }
                if (toSessionMsg.hasToServerResponse()) {
                    String requestId = sessionId + "-" + toSessionMsg.getToServerResponse().getRequestId();
                    //todo
                    listener.onToServerRpcResponse(toSessionMsg.getToServerResponse());
                }
            });
            if (sessionMetaData.getSessionType() == SessionType.SYNC) {
                deregisterSession(sessionMetaData.getSessionInfo());
            }
        } else {
            log.trace("Processing broadcast notification: {}", toSessionMsg);
        }
    }

    boolean hasSession(SessionInfoProto sessionInfo) {
        return sessions.containsKey(toSessionId(sessionInfo));
    }

    private void reportActivityInternal(SessionInfoProto sessionInfo) {
        UUID sessionId = toSessionId(sessionInfo);
        SessionActivityData sessionMetaData = sessionsActivity.computeIfAbsent(sessionId, id -> new SessionActivityData(sessionInfo));
        sessionMetaData.updateLastActivityTime();
    }

    protected UUID toSessionId(SessionInfoProto sessionInfo) {
        return new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB());
    }

    private <T> void sendToDeviceActor(SessionInfoProto sessionInfo, T message) {

    }

    private <T> void sendToRuleEngine(SessionInfoProto sessionInfo, T message, QueueCallback callback) {

    }

    /**
     * 检测不活动的session 并且上报激活
     */
    private void checkInactivityAndReportActivity() {
        long expTime = System.currentTimeMillis() - sessionInactivityTimeout;
        HashSet<UUID> sessionsToRemove = new HashSet<>();
        sessionsActivity.forEach(((sessionId, sessionActivityData) -> {
            long lastActivityTime = sessionActivityData.getLastActivityTime();
            SessionMetaData sessionMetaData = sessions.get(sessionId);
            if (sessionMetaData != null) {
                sessionActivityData.setSessionInfo(sessionMetaData.getSessionInfo());
            } else {
                sessionsToRemove.add(sessionId);
            }
            SessionInfoProto sessionInfo = sessionActivityData.getSessionInfo();
            //TODO 处理网关session

            if (lastActivityTime < expTime) {
                if (sessionMetaData != null) {
                    if (log.isDebugEnabled()) {
                        log.debug("[{}] Session has expired due to last activity time: {}", sessionId, lastActivityTime);
                    }
                }
                sessions.remove(sessionId);
                sessionsToRemove.add(sessionId);
                process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null);
                if (sessionMetaData != null) {
                    sessionMetaData.getListener().onRemoteSessionCloseCommand(sessionId, SESSION_CLOSE_NOTIFICATION_PROTO);
                }
            } else {
                if (lastActivityTime > sessionActivityData.getLastReportedActivityTime()) {
                    final long lastActivityTimeFinal = lastActivityTime;
                    SubscriptionInfoProto subscriptionInfo = SubscriptionInfoProto.newBuilder()
                            .setAttributeSubscription(sessionMetaData != null && sessionMetaData.isSubscribedToAttributes())
                            .setRpcSubscription(sessionMetaData != null && sessionMetaData.isSubscribedToRPC())
                            .setLastActivityTime(lastActivityTime)
                            .build();
                    process(sessionInfo, subscriptionInfo, new TransportServiceCallback<GetAttributeResponseMsg>() {
                        @Override
                        public void onSuccess(GetAttributeResponseMsg msg) {
                            sessionActivityData.setLastReportedActivityTime(lastActivityTimeFinal);
                        }

                        @Override
                        public void onError(Throwable e) {
                            log.error("[{}] Failed to report last activity time", sessionId, e);
                        }
                    });
                }
            }
        }));
        // Removes all closed or short-lived sessions.
        sessionsToRemove.forEach(sessionsActivity::remove);
    }

    private void invalidateRateLimits() {
        //rateLimitService.invalidateRateLimitsIpTable(sessionInactivityTimeout);
    }

    public void onDeviceProfileUpdate(DeviceProfile deviceProfile) {

        long deviceProfileId = deviceProfile.getId();
        sessions.forEach((id, md) -> {
            //如果传输类型不同, 暂时先关闭session
            if (md.getSessionInfo().getDeviceProfileId() == deviceProfileId) {
                SessionInfoProto newSessionInfo = SessionInfoProto.newBuilder()
                        .mergeFrom(md.getSessionInfo())
                        .setDeviceProfileId(deviceProfileId)
                        .setDeviceType(deviceProfile.getName())
                        .build();
                md.setSessionInfo(newSessionInfo);
                transportCallbackExecutor.submit(() -> {
                    md.getListener().onDeviceProfileUpdate(newSessionInfo, deviceProfile);
                });
            }
        });
        eventPublisher.publishEvent(new DeviceProfileUpdatedEvent(this, deviceProfile));
    }

    private void onDeviceUpdate(Device device) {

        long deviceId = device.getDeviceId();
        long deviceProfileId = device.getDeviceProfileId();
        sessions.forEach((id, md) -> {
            if (md.getSessionInfo().getDeviceId() == deviceId) {

                DeviceProfile newDeviceProfile;
                if (md.getSessionInfo().getDeviceProfileId() != deviceProfileId) {
                    //新deviceProfile
                    newDeviceProfile = new DeviceProfile();
                } else {
                    newDeviceProfile = null;
                }
                SessionInfoProto newSessionInfo = SessionInfoProto.newBuilder()
                        .mergeFrom(md.getSessionInfo())
                        .setDeviceProfileId(deviceProfileId)
                        .setDeviceId(deviceId)
                        .setDeviceName(device.getName())
                        .setDeviceType(device.getType()).build();

                //是否网关
                md.setSessionInfo(newSessionInfo);
                transportCallbackExecutor.submit(() -> {
                    md.getListener().onDeviceUpdate(newSessionInfo, device, Optional.ofNullable(newDeviceProfile));
                });
            }
        });
        eventPublisher.publishEvent(new DeviceUpdatedEvent(this, device));
    }

    private void onDeviceDeleted(Long deviceId) {

        sessions.forEach((id, md) -> {
            if (deviceId.equals(md.getSessionInfo().getDeviceId())) {
                transportCallbackExecutor.submit(() -> {
                    md.getListener().onDeviceDeleted(deviceId);
                });
            }
        });
        eventPublisher.publishEvent(new DeviceDeletedEvent(this, deviceId));
    }

    private static SessionEventMsg getSessionEventMsg(SessionEvent event) {
        return SessionEventMsg.newBuilder()
                .setSessionType(SessionType.ASYNC)
                .setEvent(event).build();
    }

    private ValidateDeviceCredentialsResponse processValidateResponseMsg(DeviceTransportType transportType, ValidateDeviceCredentialsResponseMsg responseMsg) {

        ValidateDeviceCredentialsResponse result = new ValidateDeviceCredentialsResponse();
        if (responseMsg.hasDeviceInfo()) {
            result.setCredentials(responseMsg.getCredentialsBody());
            TransportDeviceInfo deviceInfo = getTransportDeviceInfo(responseMsg.getDeviceInfo());
            result.setDeviceInfo(deviceInfo);
            ByteString profileBody = responseMsg.getProfileBody();
            if (!profileBody.isEmpty()) {
                DeviceProfile deviceProfile = transportDeviceProfileCache.getOrCreate(deviceInfo.getDeviceProfileId(), profileBody);
                if (transportType != DeviceTransportType.DEFAULT && deviceProfile != null
                        && deviceProfile.getTransportType() != DeviceTransportType.DEFAULT
                        && deviceProfile.getTransportType() != transportType) {
                    log.debug("[{}] Device profile [{}] has different transport type: {}, expected: {}",
                            deviceInfo.getDeviceId(), deviceInfo.getDeviceProfileId(), deviceProfile.getTransportType(), transportType);
                    throw new IllegalStateException("Device profile has different transport type: "
                            + deviceProfile.getTransportType() + ". Expected: " + transportType);
                }
                result.setDeviceProfile(deviceProfile);
            }
        }
        return result;
    }

    private TransportDeviceInfo getTransportDeviceInfo(DeviceInfoProto di) {
        TransportDeviceInfo tdi = new TransportDeviceInfo();
        tdi.setTenantId(TenantId.valueOf(di.getTenantId()));
        tdi.setProjectId(di.getProjectId());
        tdi.setDeviceId(di.getDeviceId());
        tdi.setDeviceProfileId(di.getDeviceProfileId());
        tdi.setAdditionalInfo(di.getAdditionalInfo());
        tdi.setDeviceName(di.getDeviceName());
        tdi.setDeviceType(di.getDeviceType());
        if (StringUtils.isNullOrEmpty(di.getPowerMode())) {
            tdi.setPowerMode(PowerMode.valueOf(di.getPowerMode()));
            tdi.setE_drxCycle(di.getEdrxCycle());
            tdi.setPsmActivityTimer(di.getPsmActivityTimer());
            tdi.setPagingTransmissionWindow(di.getPagingTransmissionWindow());
        }
        return tdi;
    }
}
